package com.github.binarylei.zookeeper.watcher;

import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;

/**
 * @author: leigang
 * @version: 2018-04-05
 */
public class ConnectionWatcher implements Watcher, Closeable {
    private static final int SESSION_TIMEOUT = 5000;
    /** 定义原子变量，用于记录watcher数 */
    private AtomicInteger seq = new AtomicInteger();

    protected ZooKeeper zk;
    private CountDownLatch connectedSignal = new CountDownLatch(1);

    public void connect(String host) {
        this.close();
        try {
            this.zk = new ZooKeeper(host, SESSION_TIMEOUT, this);
            this.connectedSignal.await();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (event == null) {
            return;
        }
        // 连接状态
        KeeperState keeperState = event.getState();
        // 事件类型
        EventType eventType = event.getType();
        // 受影响的path
        String path = event.getPath();
        String logPrefix = "【Watcher-" + seq.incrementAndGet() + "】";

        if(keeperState == KeeperState.SyncConnected) {
            // 成功连接上服务器
            if (eventType == EventType.None) {
                System.out.println(logPrefix + "成功连接上服务器");
                this.connectedSignal.countDown();
            }
            // 节点创建
            else if (eventType == EventType.NodeCreated) {
                System.out.println(logPrefix + "节点创建：" + path);
            }
            // 节点数据更新
            else if (eventType == EventType.NodeDataChanged) {
                System.out.println(logPrefix + "节点数据更新："+ path);
            }
            // 节点删除
            else if (eventType == EventType.NodeDeleted) {
                System.out.println(logPrefix + "节点删除：" + path);
            }
        } else if (keeperState == KeeperState.Disconnected) {
            System.out.println(logPrefix + "连接断开");
        } else if (keeperState == KeeperState.AuthFailed) {
            System.out.println(logPrefix + "权限认证失败");
        } else if (keeperState == KeeperState.Expired) {
            System.out.println(logPrefix + "连接过期");
        }
    }

    // 关闭连接
    public void close() {
        if (this.zk != null) {
            try {
                this.zk.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public ZooKeeper getZk() {
        return zk;
    }

    public void setZk(ZooKeeper zk) {
        this.zk = zk;
    }
}
